using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Factory;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service.Template
{

    public abstract class SuperCfTemplate<K, SN, N> : AbstractColumnFamilyTemplate<K, SN>
	{
        public SuperCfTemplate(Keyspace keyspace, String columnFamily,
            Serializer<K> keySerializer, Serializer<SN> topSerializer,
            Serializer<N> subSerializer)
            : base(keyspace, columnFamily, keySerializer, topSerializer)
        {
            SubSerializer = subSerializer;
        }
        public Serializer<N> SubSerializer
        {
            get;
            set;
        }
        public bool IsColumnsExist(K key)
        {
            return CountColumns(key) > 0;
        }
        public bool IsSubColumnsExist(K key, SN superColumnName)
        {
            return CountSubColumns(key, superColumnName) > 0;
        }
        public int CountColumns(K key)
        {
            return CountColumns(key, (SN)ALL_COLUMNS_START, (SN)ALL_COLUMNS_END,
                ALL_COLUMNS_COUNT);
        }
        public int CountSubColumns(K key, SN superColumnName)
        {
            return CountSubColumns(key, superColumnName, (N)ALL_COLUMNS_START,
                (N)ALL_COLUMNS_END, ALL_COLUMNS_COUNT);
        }
        public int CountSubColumns(K key, SN superColumnName, N start,
            N end, int max)
        {
            SubCountQuery<K, SN, N> query = HFactory.CreateSubCountQuery(
                Keyspace, KeySerializer, TopSerializer, SubSerializer);
            query.SetKey(key);
            query.SetColumnFamily(ColumnFamily);
            query.SetSuperColumn(superColumnName);
            query.SetRange(start, end, max);
            return query.Execute().Get();
        }
        public int CountColumns(K key, SN start, SN end, int max)
        {
            SuperCountQuery<K, SN> query = HFactory.CreateSuperCountQuery(Keyspace,
                KeySerializer, TopSerializer);
            query.SetKey(key);
            query.SetColumnFamily(ColumnFamily);
            query.SetRange(start, end, max);
            return query.Execute().Get();
        }
        public HColumn<N, V> QuerySingleSubColumn<V>(K key,
            SN columnName, N subColumnName, Serializer<V> valueSerializer)
        {
            SuperCfResult<K,SN,N> result = DoExecuteSlice(key, columnName, ActiveSlicePredicate);
            if (result == null)
            {
                return null;
            }
            HColumn<N,byte[]> origCol = result.GetColumn(subColumnName);
            if ( columnName == null || origCol == null )
            {
                return null;
            }
            return new HColumnImpl<N, V>(subColumnName, 
                valueSerializer.FromBytes(origCol.Value), origCol.Clock, 
                SubSerializer, valueSerializer);
        }
        public SuperCfResult<K, SN, N> querySuperColumns(K key, HSlicePredicate<SN> predicate)
        {
            return DoExecuteSlice(key,default(SN),predicate);
        }
        public SuperCfResult<K, SN, N> querySuperColumns(K key, List<SN> sColumnNames)
        {
            HSlicePredicate<SN> workingSlicePredicate = new HSlicePredicate<SN>(TopSerializer);
            workingSlicePredicate.SetColumnNames(sColumnNames);
            return DoExecuteSlice(key, default(SN), workingSlicePredicate);
        }
        public SuperCfResult<K, SN, N> querySuperColumns(List<K> keys, List<SN> sColumnNames)
        {
            HSlicePredicate<SN> workingSlicePredicate = new HSlicePredicate<SN>(TopSerializer);
            workingSlicePredicate.SetColumnNames(sColumnNames);
            return DoExecuteMultigetSlice(keys, workingSlicePredicate);
        }
        public SuperCfResult<K, SN, N> querySuperColumns(List<K> keys) 
        {
            return DoExecuteMultigetSlice(keys, ActiveSlicePredicate);
        }
        public T querySuperColumns<T>(K key, List<SN> sColumnNames,
            SuperCfRowMapper<K, SN, N, T> mapper)
        {
            HSlicePredicate<SN> workingSlicePredicate = new HSlicePredicate<SN>(TopSerializer);
            workingSlicePredicate.SetColumnNames(sColumnNames);
            return mapper.MapRow(DoExecuteSlice(key, default(SN), workingSlicePredicate));
        }
        public SuperCfResult<K, SN, N> querySuperColumns(K key)
        {
            return DoExecuteSlice(key, default(SN), ActiveSlicePredicate);
        }

        public SuperCfResult<K, SN, N> QuerySuperColumn(K key, SN sColumnName)
        {
            HSlicePredicate<SN> workingSlicePredicate = new HSlicePredicate<SN>(TopSerializer);
            workingSlicePredicate.AddColumnName(sColumnName);
            return DoExecuteSlice(key, sColumnName, workingSlicePredicate);
        }

        public SuperCfUpdater<K, SN, N> CreateUpdater(K key, SN sColumnName)
        {
            return CreateUpdater(key).AddSuperColumn(sColumnName);
        }

        public SuperCfUpdater<K, SN, N> CreateUpdater(K key)
        {
            SuperCfUpdater<K, SN, N> updater = new SuperCfUpdater<K, SN, N>(this, ColumnFactory);
            updater.AddKey(key);
            return updater;
        }
        public void Update(SuperCfUpdater<K, SN, N> updater)
        {
            updater.UpdateInternal();
            updater.Update();
            ExecuteIfNotBatched(updater.CurrentMutator);
        }
        public new void DeleteColumn(K key, SN sColumnName)
        {
            CreateMutator().SuperDelete(key, ColumnFamily, sColumnName, TopSerializer);
        }
        public void DeleteRow(K key)
        {
            CreateMutator().Delete(key, ColumnFamily, default(N), null);
        }
        protected abstract SuperCfResult<K,SN,N> DoExecuteSlice(K key, SN sColumnName, HSlicePredicate<SN> predicate);
        protected abstract SuperCfResult<K,SN,N> DoExecuteMultigetSlice(List<K> keys, HSlicePredicate<SN> predicate);
	}
}